/****

    activequant - activestocks.eu

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License along
    with this program; if not, write to the Free Software Foundation, Inc.,
    51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

	
	contact  : contact@activestocks.eu
    homepage : http://www.activestocks.eu

****/
package org.activequant.data.util.converter;

import org.activequant.core.domainmodel.InstrumentSpecification;
import org.activequant.core.domainmodel.data.Candle;
import org.activequant.core.domainmodel.data.TradeIndication;
import org.activequant.core.types.TimeFrame;
import org.activequant.data.retrieval.ICandleSubscriptionSource;
import org.activequant.data.retrieval.ISubscription;
import org.activequant.data.retrieval.ITradeIndicationSubscriptionSource;
import org.activequant.data.retrieval.SubscriptionSourceBase;
import org.activequant.util.pattern.events.IEventListener;

/**
 * Converts a candle realtime feed to trade indication realtime feed
 * by faking the price move from Open to Close. The generation
 * of trade indication is controlled by the strategy {@link #getConversion() conversion}.
 * <br>
 * CandleToTradeIndicationConverter extends SubscriptionSourceBase&lt;TradeIndication&gt; implements ITradeIndicationSubscriptionSource.<br/>
 * Defines a private CandleSubscription class. Holds the following associated variables:
 * <ul>
 * <li>candleSource(ICandleSubscriptionSource)</li>
 * <li>timeFrame(TimeFrame)</li>
 * <li>conversion(ICandleToTradeIndicationConversion)</li>
 * </ul>
 * Holds the following inherited variables:
 * <ul>
 * <li>activeBackends(Map&lt;Topic, Subscription&gt;)</li>
 * <li>activeSubscriptions(Queue&lt;SubscriptionDelegate&gt;)</li>
 * </ul>
 * <b>History:</b><br>
 *  - [10.07.2007] Created (Ulrich Staudinger)<br>
 *  - [05.08.2007] Cleanup (Erik Nijkamp)<br>
 *  - [29.09.2007] cleanup + moved to new domain model (Erik Nijkamp)<br>
 *  - [29.09.2007] Refactored push/pull approach (Erik Nijkamp)<br>
 *  - [09.11.2007] Moved to subscription data model (Mike Kroutikov)<br>
 *  - [09.30.2008] Isolated generation code following a strategy pattern (Mike Kroutikov)<br>
 *
 *  @author Ulrich Staudinger
 */
public class CandleToTradeIndicationConverter extends SubscriptionSourceBase<TradeIndication> implements ITradeIndicationSubscriptionSource {
	/**
	 * private final ICandleSubscriptionSource candleSource; 
	 */
	private final ICandleSubscriptionSource candleSource; 
	/**
	 * private final TimeFrame timeFrame;
	 */
	private final TimeFrame timeFrame;
	
	/**
	 * Creates new converter from candle subscription source and time frame.<br/>
	 * constructs a CandleToTradeIndicationConverter (extends SubscriptionSourceBase&lt;TradeIndication&gt; implements ITradeIndicationSubscriptionSource)
	 * using the given candleSource(ICandleSubscriptionSource) and timeFrame(TimeFrame) to set its associated candleSource(ICandleSubscriptionSource) and timeFrame(TimeFrame)
	 * @param candleSource source of candle subscription.
	 * @param timeFrame determines what time frame to request from candle subscription source.
	 */
	public CandleToTradeIndicationConverter(ICandleSubscriptionSource candleSource, TimeFrame timeFrame) {
		super();
		this.candleSource = candleSource;
		this.timeFrame = timeFrame;
	}
	/**
	 * private ICandleToTradeIndicationConversion conversion = new LinearCandleToTradeIndicationConversion();
	 */
	private ICandleToTradeIndicationConversion conversion = new LinearCandleToTradeIndicationConversion();

	/**
	 * Determines how Candles are converted to a list of TradeIndications.<br/>
	 * returns the associated conversion(ICandleToTradeIndicationConversion)
	 * @return conversion strategy.
	 */
	public ICandleToTradeIndicationConversion getConversion() {
		return conversion;
	}

	/**
	 * Conversion strategy.<br/>
	 * sets the associated conversion(ICandleToTradeIndicationConversion) with the given val(ICandleToTradeIndicationConversion)
	 * @param conversion strategy.
	 */
	public void setConversion(ICandleToTradeIndicationConversion val) {
		conversion = val;
	}
	/**
	 * A private class. Subscription subclass. Holds the following associated variables:
	 * <ul>
	 * <li>subscription(ISubscription&lt;Candle&gt;)</li>
	 * <li>listener(IEventListener&lt;Candle&gt;)</li>
	 * </ul>
	 * Holds the following inherited variables:
	 * <ul>
	 * <li>isActive(AtomicBoolean)</li>
	 * <li>refCount(AtomicInteger)</li>
	 * <li>instrument(InstrumentSpecification)</li>
	 * <li>event(Event&lt;T&gt;)</li>
	 * </ul>
	 * @author Dimitar
	 *
	 */
	private class CandleSubscription extends Subscription {
		/**
		 * private final ISubscription&lt;Candle&gt; subscription;
		 */
		private final ISubscription<Candle> subscription;
		/**
		 * private final IEventListener&lt;Candle&gt; listener = new IEventListener&lt;Candle&gt;(){...}<br/>
		 * A listener whose job is to convert each received candle(Candle) event into a series of TradeIndications (by using the conversion(ICandleToTradeIndicationConversion))
		 * and then fire each one as an event on the inherited event(Event&lt;TradeIndication&gt;)
		 */
		private final IEventListener<Candle> listener = new IEventListener<Candle>() {
			public void eventFired(Candle candle) {
				for(TradeIndication ti : conversion.convert(candle)) {
					try {
						fireEvent(ti);
					} catch(Exception ex) {
						log.error(ex);
						ex.printStackTrace();
					}
				}
			}
		};
		/**
		 * constructs a CandleSubscription (extends Subscription) for the given spec(InstrumentSpecification)<br/>
		 * Sets its associated subscription(ISubscription&lt;Candle&gt;) with a new Candle subscription for the given spec(InstrumentSpecification) and timeFrame(TimeFrame) using the 
		 * candleSource(ICandleSubscriptionSource).
		 * @param spec
		 * @throws Exception
		 */
		public CandleSubscription(InstrumentSpecification spec) throws Exception {
			subscription = candleSource.subscribe(spec, timeFrame);
		}
		/**
		 * registers the associated listener(IEventListener&lt;Candle&gt;) to listen for Candle events from the associated subscription(ISubscription&lt;Candle&gt;)
		 * and then activates the same associated subscription(ISubscription&lt;Candle&gt;).
		 */
		@Override
		protected void handleActivate() {
			subscription.addEventListener(listener);
			subscription.activate();
		}
		/**
		 * Cancels the associated subscription(ISubscription&lt;Candle&gt;) and then stops the associated listener(IEventListener&lt;Candle&gt;) from listening to it.
		 */
		@Override
		protected void handleCancel() {
			subscription.cancel();
			subscription.removeEventListener(listener);
		}
	}
	/**
	 * Creates and returns a CandleSubscription (extends Subscription) for the given spec(InstrumentSpecification)<br/>
	 * Sets its associated subscription(ISubscription&lt;Candle&gt;) with a new Candle subscription for the given spec(InstrumentSpecification) and associated timeFrame(TimeFrame) using the 
	 * associated candleSource(ICandleSubscriptionSource).
	 */
	@Override
	protected CandleSubscription createSubscription(InstrumentSpecification spec, TimeFrame timeFrame) {
		try {
			return new CandleSubscription(spec);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
	/**
	 * Returns/creates a SubscriptionDelegate(ISubscription&lt;TradeIndication&gt;) for the given spec(InstrumentSpecification) and TIMEFRAME_1_TICK.
	 */
	public ISubscription<TradeIndication> subscribe(InstrumentSpecification spec) throws Exception {
		return super.subscribe(spec, TimeFrame.TIMEFRAME_1_TICK);
	}
}